// Copyright (c) 2021 Presto Labs Pte. Ltd.
// Author: leon

#include "coin2/exchange/gopax/feed_v1/processor.h"

#include <string>
#include <utility>

#include "coin2/base/conversion.h"
#include "coin2/exchange/base/api_util/json_schema.h"
#include "coin2/exchange/gopax/feed_v1/update.h"
#include "coin2/exchange/util/json_util.h"

namespace coin2::exchange::gopax::feed_v1::impl {

void GopaxFeedProcessor::OnTopicRecord(const base::executor::TopicRecordPacket& packet) {
  upd_mgr_.onBeginPacket(packet);

  rapidjson::Document document;
  document.Parse<rapidjson::kParseNumbersAsStringsFlag>(packet.payload.data());
  if (document.HasParseError()) {
    THROW() << "JSON parse error: " << GetParseError_En(document.GetParseError());
  }

  if (parser_config_.validate_schema() && schema_) {
    schema_->AcceptOrThrow(document);
  }

  if (document.IsObject() &&
      document.HasMember("n") &&
      document.HasMember("o")) {
    const auto& channel = GopaxChannel::Parse(document["n"].GetString());
    if (!CheckChannelFilter(channel.subscribe_channel)) {
      upd_mgr_.onEndPacket(packet);
      return;
    }

    if (!document["o"].HasMember("tradingPairName")) {
      upd_mgr_.onEndPacket(packet);
      return;
    }
    const auto& native_symbol = document["o"]["tradingPairName"].GetString();
    if (channel.event_type == "PublicTradeEvent") {
      const auto& transaction_data = document["o"];
      if (!upd_mgr_.is_symbol_registered(native_symbol)) {
        return;
      }
      Trade* trade = upd_mgr_.GetTradeMutable();
      GopaxFeedUpdateParser::ParseTrade(transaction_data, trade);
      upd_mgr_.PublishTrade(native_symbol, false);
    } else if (channel.event_type == "OrderBookEvent" ||
               channel.event_type == "SubscribeToTradingPair") {
      const auto& orderbook_data = document["o"];
      auto* book = upd_mgr_.GetBookMutable(native_symbol);
      if (book != nullptr) {
        GopaxFeedUpdateParser::ParseBook(orderbook_data, book);
        upd_mgr_.PublishBook();
      }
    } else {
      LOG(INFO) << "[Gopax Feed Processor] unknown msg for " << packet.topic_name << ": "
                << packet.payload;
    }
  } else {
    // meta msg like subscription response...
    LOG(INFO) << "[Gopax Feed Processor] meta msg for " << packet.topic_name << ": "
              << packet.payload;
  }

  upd_mgr_.onEndPacket(packet);
}
}  // namespace coin2::exchange::gopax::feed_v1::impl
